package spark_graphx.spark_graphx

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.Edge
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.graphx.Graph
import org.apache.spark.graphx._

object IdIpImeiGraphX {
  def ipToLong(strIp: String): Long = {
    val ip = strIp.split("\\.");
    if (ip.length != 4) return 0;
    return (ip(0).toLong << 24) + (ip(1).toLong << 16) + (ip(2).toLong << 8) + ip(3).toLong
  }

  def longToIP(longIp: Long): String = {
    var sb = new StringBuffer("");
    // 直接右移24位  
    sb.append(String.valueOf((longIp >>> 24)));
    sb.append(".");
    // 将高8位置0，然后右移16位  
    sb.append(String.valueOf((longIp & 0x00FFFFFF) >>> 16));
    sb.append(".");
    // 将高16位置0，然后右移8位  
    sb.append(String.valueOf((longIp & 0x0000FFFF) >>> 8));
    sb.append(".");
    // 将高24位置0  
    sb.append(String.valueOf((longIp & 0x000000FF)));
    return sb.toString();
  }

  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:/worktools/hadoop-2.7.3");

    //设置运行环境
    val conf = new SparkConf().setAppName("IdIpImeiGraphX").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val user_imei_RDD: RDD[String] = sc.textFile("file:///D:/文档/graphx/data/user-imei.txt", 2);
    val user_ip_RDD: RDD[String] = sc.textFile("file:///D:/文档/graphx/data/user-ip.txt", 2);
    val user_pay_RDD: RDD[String] = sc.textFile("file:///D:/文档/graphx/data/user-pay-id.txt", 2);
    val driver_imei_RDD: RDD[String] = sc.textFile("file:///D:/文档/graphx/data/driver-imei.txt", 2);
    val driver_ip_RDD: RDD[String] = sc.textFile("file:///D:/文档/graphx/data/driver-ip.txt", 2);

    var user_imei_edges: RDD[Edge[String]] = user_imei_RDD.map(line => {
      val fields: Array[String] = line.split("\t", -1)
      Edge(("1" + fields(0)).toLong, ("3" + fields(1)).toLong, fields(2))
    });

    var user_ip_edges: RDD[Edge[String]] = user_ip_RDD.map(line => {
      val fields: Array[String] = line.split("\t", -1)
      Edge(("1" + fields(0)).toLong, ("4" + ipToLong(fields(1))).toLong, fields(1))
    });

    var user_pay_edges: RDD[Edge[String]] = user_pay_RDD.map(line => {
      val fields: Array[String] = line.split("\t", -1)
      Edge(("1" + fields(0)).toLong, ("5" + fields(1)).toLong, fields(2))
    });

    var driver_imei_edges: RDD[Edge[String]] = driver_imei_RDD.map(line => {
      val fields: Array[String] = line.split("\t", -1)
      Edge(("2" + fields(0)).toLong, ("3" + fields(1)).toLong, fields(2))
    });
    
    var driver_ip_edges: RDD[Edge[String]] = driver_ip_RDD.map(line => {
      val fields: Array[String] = line.split("\t", -1)
      Edge(("2" + fields(0)).toLong, ("4" + ipToLong(fields(1))).toLong, fields(1))
    });

    var edgeRDD = user_imei_edges.union(user_ip_edges).union(user_pay_edges).union(driver_imei_edges).union(driver_ip_edges);
    //构造图Graph[VD,ED]
    val graph: Graph[VertexId, String] = Graph.fromEdges(edgeRDD, 1, StorageLevel.MEMORY_AND_DISK_SER,  StorageLevel.MEMORY_AND_DISK_SER)
        //Degrees操作
    println("找出图中最大的出度、入度、度数：")
    def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
      if (a._2 > b._2) a else b
    }
    
    println("max of outDegrees:" + graph.outDegrees.reduce(max) + " max of inDegrees:" + graph.inDegrees.reduce(max) + " max of Degrees:" + graph.degrees.reduce(max))
    println    
    
    //      //边操作：找出图中amount属性大于50的边
    //      println("找出图中amount属性大于50的边：")
    //  //    graph.edges.filter(e => e.attr.amount > 50).collect().foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))
    //      graph.edges.filter(e => e.attr.amount > 50).saveAsTextFile("file:///D:/文档/graphx/amount-over-50.txt")
    //      println
    //      
    //      
    //      val edgeRDD : RDD[((VertexId,VertexId),Long)]  = graph.edges.map { e => ((e.srcId, e.dstId),1) }
    //  //    edgeRDD.reduceByKey { (v1, v2) => v1+v2 }.filter(t => t._2 >1 ).foreach( e => println(s"${e._1} count ${e._2}") );
    //      edgeRDD.reduceByKey { (v1, v2) => v1+v2 }.filter(t => t._2 >1 ).saveAsTextFile("file:///D:/文档/graphx/cc-over-2.txt");

  }

}